package com.study.chapter09;

import com.study.chapter05.source.ClickSource;
import com.study.entity.Action;
import com.study.entity.Event;
import com.study.entity.Pattern;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/17 20:22
 */
public class BehaviorPatternExampleTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //用户的行为数据流
        DataStreamSource<Action> actionStream = env.fromElements(
                new Action("Bob", "login"),
                new Action("Bob", "pay"),
                new Action("Alice", "login"),
                new Action("Alice", "order")
        );
        //行为模式流，基于它构建广播流
        DataStreamSource<Pattern> patternStream = env.fromElements(
                new Pattern("login", "pay"),
                new Pattern("login", "order")
        );

        //定义广播状态描述器
        MapStateDescriptor<Void, Pattern> descriptor = new MapStateDescriptor<>("pattern", Types.VOID, Types.POJO(Pattern.class));
        BroadcastStream<Pattern> broadcastStream = patternStream.broadcast(descriptor);

        //连接两条流进行处理
        actionStream.keyBy(act -> act.userId)
                .connect(broadcastStream)
                .process(new PatternDetector())
                .print();

        env.execute();
    }

    public static class PatternDetector extends KeyedBroadcastProcessFunction<String,Action,Pattern, Tuple2<String,Pattern>>{
        // 定义一个值状态，保存上一次用户行为
        private ValueState<String> lastActionState;

        @Override
        public void open(Configuration parameters) throws Exception {
            lastActionState = getRuntimeContext().getState(new ValueStateDescriptor<String>("last-action",String.class));
        }

        @Override
        public void processElement(Action value, ReadOnlyContext ctx, Collector<Tuple2<String, Pattern>> out) throws Exception {
            //从广播状态中获取匹配模式
            ReadOnlyBroadcastState<Void, Pattern> patternState = ctx.getBroadcastState(new MapStateDescriptor<>("pattern", Types.VOID, Types.POJO(Pattern.class)));
            Pattern pattern = patternState.get(null);

            //获取用户上一次的行为
            String lastAction = lastActionState.value();
            if (pattern != null && lastAction != null){
                if (StringUtils.equals(pattern.action1,lastAction) && StringUtils.equals(pattern.action2,value.action)){
                    out.collect(new Tuple2<>(ctx.getCurrentKey(),pattern));
                }
            }
            //更新状态
            lastActionState.update(value.action);
        }

        @Override
        public void processBroadcastElement(Pattern value, Context ctx, Collector<Tuple2<String, Pattern>> out) throws Exception {
            //从上下文中获取广播状态，并用当前数据更新状态
            BroadcastState<Void, Pattern> patternState = ctx.getBroadcastState(new MapStateDescriptor<>("pattern", Types.VOID, Types.POJO(Pattern.class)));
            // 将广播状态更新为当前的pattern
            patternState.put(null,value);
        }
    }

}
